/*
 * Copyright 1999-2011 Alibaba Group.
 *  
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.dubbo.rpc.protocol.dubbo;

import static com.alibaba.dubbo.rpc.protocol.dubbo.CallbackServiceCodec.encodeInvocationArgument;

import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.HeaderProvider2;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.VersionSupport;
import com.alibaba.dubbo.common.io.Bytes;
import com.alibaba.dubbo.common.io.UnsafeByteArrayInputStream;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.serialize.ObjectInput;
import com.alibaba.dubbo.common.serialize.ObjectOutput;
import com.alibaba.dubbo.common.serialize.Serialization;
import com.alibaba.dubbo.common.utils.ReflectUtils;
import com.alibaba.dubbo.common.utils.StringUtils;
import com.alibaba.dubbo.remoting.Channel;
import com.alibaba.dubbo.remoting.Codec2;
import com.alibaba.dubbo.remoting.buffer.ChannelBuffer;
import com.alibaba.dubbo.remoting.exchange.Request;
import com.alibaba.dubbo.remoting.exchange.Response;
import com.alibaba.dubbo.remoting.exchange.codec.ExchangeCodec;
import com.alibaba.dubbo.remoting.transport.CodecSupport;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Result;
import com.alibaba.dubbo.rpc.RpcInvocation;

/**
 * Dubbo codec.
 *
 * @author qianlei
 * @author chao.liuc
 */
public class DubboCodec extends ExchangeCodec implements Codec2 {

	private static final Logger log = LoggerFactory.getLogger(DubboCodec.class);

	public static final String NAME = "dubbo";

	public static final String DUBBO_VERSION = VersionSupport.getVersion(
			DubboCodec.class, VersionSupport.getVersion());

	public static final byte RESPONSE_WITH_EXCEPTION = 0;

	public static final byte RESPONSE_VALUE = 1;

	public static final byte RESPONSE_NULL_VALUE = 2;

	public static final Object[] EMPTY_OBJECT_ARRAY = new Object[0];

	public static final Class<?>[] EMPTY_CLASS_ARRAY = new Class<?>[0];

	public static final String CLASS_DESC_AUTO = "auto";

	@Override
	protected Object decodeBody(Channel channel, InputStream is, byte[] header,
			Map<String, String> headers,boolean new_protocol) throws IOException {
		byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
		Serialization s = CodecSupport.getSerialization(channel.getUrl(), proto);
		// get request id.
		long id = Bytes.bytes2long(header, 4);
		if ((flag & FLAG_REQUEST) == 0) {
			// decode response.
			Response res = new Response(id);
			res.setNewProtocol(new_protocol);
			if ((flag & FLAG_EVENT) != 0) {
				res.setEvent(Response.HEARTBEAT_EVENT);
			}
			// get status.
			byte status = header[3];
			res.setStatus(status);
			if (status == Response.OK) {
				try {
					Object data;
					if (res.isHeartbeat()) {
						data = decodeHeartbeatData(channel,
								deserialize(s, channel.getUrl(), is));
					} else if (res.isEvent()) {
						data = decodeEventData(channel,
								deserialize(s, channel.getUrl(), is));
					} else {
						DecodeableRpcResult result;
						if (channel.getUrl().getParameter(
								Constants.DECODE_IN_IO_THREAD_KEY,
								Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
							result = new DecodeableRpcResult(channel, res, is,
									(Invocation) getRequestData(id), proto);
							result.decode();
						} else {
							result = new DecodeableRpcResult(channel, res,
									new UnsafeByteArrayInputStream(
											readMessageData(is)),
									(Invocation) getRequestData(id), proto);
						}
						result.addHeaders(headers);
						data = result;
					}
					res.setResult(data);
				} catch (Throwable t) {
					if (log.isWarnEnabled()) {
						log.warn("Decode response failed: " + t.getMessage(), t);
					}
					res.setStatus(Response.CLIENT_ERROR);
					res.setErrorMessage(StringUtils.toString(t));
				}
			} else {
				res.setErrorMessage(deserialize(s, channel.getUrl(), is)
						.readUTF());
			}
			return res;
		} else {
			// decode request.
			Request req = new Request(id);
			req.setNewProtocol(new_protocol);
			req.setVersion("2.0.0");
			req.setTwoWay((flag & FLAG_TWOWAY) != 0);
			if ((flag & FLAG_EVENT) != 0) {
				req.setEvent(Request.HEARTBEAT_EVENT);
			}
			try {
				Object data;
				if (req.isHeartbeat()) {
					data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is));
				} else if (req.isEvent()) {
					data = decodeEventData(channel, deserialize(s, channel.getUrl(), is));
				} else {
					DecodeableRpcInvocation inv;
					if (channel.getUrl().getParameter(
							Constants.DECODE_IN_IO_THREAD_KEY,
							Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
						inv = new DecodeableRpcInvocation(channel, req, is,
								proto);
						inv.decode();
					} else {
						inv = new DecodeableRpcInvocation(channel, req,
								new UnsafeByteArrayInputStream(
										readMessageData(is)), proto);
					}
					inv.addHeaders(headers);
					data = inv;
				}
				req.setData(data);
			} catch (Throwable t) {
				if (log.isWarnEnabled()) {
					log.warn("Decode request failed: " + t.getMessage(), t);
				}
				// bad request
				req.setBroken(true);
				req.setData(t);
			}
			return req;
		}
	}

	private ObjectInput deserialize(Serialization serialization, URL url,
			InputStream is) throws IOException {
		return serialization.deserialize(url, is);
	}

	private byte[] readMessageData(InputStream is) throws IOException {
		if (is.available() > 0) {
			byte[] result = new byte[is.available()];
			is.read(result);
			return result;
		}
		return new byte[] {};
	}

	
	@SuppressWarnings("serial")
	static private Set<String> discardKeys = new HashSet<String>(){
		{
			this.add(Constants.DUBBO_VERSION_KEY);
			this.add(Constants.PATH_KEY);
			this.add(Constants.VERSION_KEY);
			this.add(Constants.INPUT_KEY);
			this.add(Constants.OUTPUT_KEY);

		}
	};
	private Map<String, String> newHeaders(Map<String, String> att){
		Map<String,String> result = new HashMap<String,String>();
		if(att != null && !att.isEmpty()){
			for(Map.Entry<String, String> entry : att.entrySet()){
				if(discardKeys.contains(entry.getKey())) continue;
				if(entry.getValue() == null || entry.getValue().length() == 0) continue;
				result.put(entry.getKey(), entry.getValue());
			}
		}
		return result;
	}
	
	@SuppressWarnings("unused")
	private void clearHeaders(Map<String, String> att) {
		for(String s : discardKeys){
			att.remove(s);
		}
	}

	@Override
	protected int encodeRequestHeaders(Channel channel,
			ChannelBuffer buffer, int writerIndex, Serialization serialization,
			Object data) throws IOException {
		if (data instanceof RpcInvocation) {
			RpcInvocation inv = (RpcInvocation) data;
			// why not remove special key? aruan 2014.09.14
			// 写入扩展信息
			Map<String, String> att = newHeaders(
					inv.getHeaders());
			return encodeHeaders(channel, buffer, writerIndex,
					serialization, att);
		}
		return 0;
	}

	protected Map<String,String> getRequestHeaders(Request req) {
		return getHeaders(req.getData());
	}
	
	private Map<String, String> getHeaders(Object value){
		if(value instanceof HeaderProvider2) {
			return ((HeaderProvider2)value).getHeaders();
		}
		return null;
	}

	protected Map<String, String> getResponseHeaders(Response resp) {
		return getHeaders(resp.getResult());
	}

	@Override
	protected void encodeRequestData(Channel channel, ObjectOutput out,
			Object data) throws IOException {
		RpcInvocation inv = (RpcInvocation) data;

		out.writeUTF(inv.getAttachment(Constants.DUBBO_VERSION_KEY,
				DUBBO_VERSION));
		out.writeUTF(inv.getAttachment(Constants.PATH_KEY));
		out.writeUTF(inv.getAttachment(Constants.VERSION_KEY));
		
		// 写入方法名
		out.writeUTF(inv.getMethodName());
		// 写入参数类型
		Object[] args = inv.getArguments(); 
		out.writeInt(/*ReflectUtils.getDesc(inv.getParameterTypes())*/args == null?0:args.length);
		// aruan 2014.09.15 尝试写入auto/null
		//out.writeUTF(null);
		
		// 写入参数值
		if (args != null){
			for (int i = 0; i < args.length; i++) {
				out.writeObject(encodeInvocationArgument(channel, inv, i));
			}
		}
		out.writeObject(inv.getAttachments());
	}

	@Override
	protected int encodeResponseHeaders(Channel channel,
			ChannelBuffer buffer, int writerIndex, Serialization serialization,
			Object data) throws IOException {
		if (data instanceof HeaderProvider2) {
			HeaderProvider2 result = (HeaderProvider2) data;
			// why not remove special key? aruan 2014.09.14
			// 写入扩展信息
			Map<String, String> att = newHeaders(
					result.getHeaders());

			return encodeHeaders(channel, buffer, writerIndex,
					serialization, att);
		}
		return 0;
	}

	@Override
	protected void encodeResponseData(Channel channel, ObjectOutput out,
			Object data) throws IOException {
		if(data == null || !(data instanceof Result)) return;
 		Result result = (Result) data;
		Throwable th = result.getException();
		if (th == null) {
			Object ret = result.getValue();
			if (ret == null) {
				out.writeByte(RESPONSE_NULL_VALUE);
			} else {
				out.writeByte(RESPONSE_VALUE);
				out.writeObject(result.getValue());
			}
		} else {
			out.writeByte(RESPONSE_WITH_EXCEPTION);
			out.writeObject(th);
		}
	}

	
	public static final String SERIALIZER_KEY = "serializer";
	
	
	@Override
	protected Serialization getSerialization(Channel channel, byte flag,
			Map<String, String> headers) {
		if(headers != null && !headers.isEmpty()){
			String skey = headers.get(SERIALIZER_KEY);
			if(skey != null && skey.length() > 0){
				Serialization s = CodecSupport.getSerializationByName(skey);
				if(s != null){
					return s;
				}
			}
		}
		return super.getSerialization(channel, flag, headers);
	}
	

	private Serialization getSerialization(	Map<String, String> headers) {
		if(headers != null && !headers.isEmpty()){
			String skey = headers.get(SERIALIZER_KEY);
			if(skey != null && skey.length() > 0){
				Serialization s = CodecSupport.getSerializationByName(skey);
				if(s != null){
					return s;
				}
			}
		}
		return null;
	}
	
	private Serialization getSerialization(	HeaderProvider2 provider) {
		if(provider != null ){
			return getSerialization(provider.getHeaders());
		}
		return null;
	}

	@Override
	protected Serialization getSerialization(Channel channel, Request req) {
		if (req.getData() != null && req.getData() instanceof HeaderProvider2) {
			HeaderProvider2 result = (HeaderProvider2) req.getData();
			Serialization s = this.getSerialization(result);
			if(s != null){
				return s;
			}
		}
		return super.getSerialization(channel, req);
	}

	@Override
	protected Serialization getSerialization(Channel channel, Response res) {
		if (res.getResult() != null && res.getResult() instanceof HeaderProvider2) {
			HeaderProvider2 result = (HeaderProvider2) res.getResult();
			Serialization s = this.getSerialization(result);
			if(s != null){
				return s;
			}
		}
		return super.getSerialization(channel, res);
	}
}